package org.egomsl.mw.segment;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.egomsl.mw.block.BlockScanner;
import org.egomsl.mw.record.Record;
import org.egomsl.mw.record.RecordScanner;

import java.io.IOException;

public class SegmentRecordScanner implements RecordScanner {
    protected BlockScanner blockScanner;
    protected RecordScanner recordScanner;

    public SegmentRecordScanner(FileSystem fs, String path) throws IOException {
        SegmentBlockReader reader = new SegmentBlockReader(fs);

        blockScanner = reader.getBlockScanner(new Path(path));
        recordScanner = null;
    }

    public SegmentRecordScanner(FileSystem fs, Path path) throws IOException {
        SegmentBlockReader reader = new SegmentBlockReader(fs);

        blockScanner = reader.getBlockScanner(path);
        recordScanner = null;
    }


    @Override
    public boolean hashNext() {
        if(recordScanner != null && recordScanner.hashNext()) return true;
        else if(blockScanner.hasNext()) {
            recordScanner = blockScanner.next().getScanner();
            if(recordScanner.hashNext()) return true;
        }
        return false;
    }

    @Override
    public Record next() {
        return recordScanner.next();
    }

    public void close() {
        blockScanner.close();
    }
}
